﻿using Castle.Core.Logging;
using Inovout.Cloud.Data.Blob.Models;
using Inovout.Runtime;
using Microsoft.Hadoop.WebHDFS;
using Newtonsoft.Json.Linq;
using System;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Inovout.Cloud.Data.Blob.Service
{
    public static class WebHDFSClientExtension
    {
        private static string GetUri(WebHDFSClient client, File file)
        {
            string uri = client.BaseUri.ToString();
            if (!uri.EndsWith("/"))
            {
                uri += "/";
            }
            uri += "webhdfs/v1" + file.Path + "?user.name=hdfs";// +  file.Owner;
            return uri;
        }

        public static async Task<string> CreateFile(this WebHDFSClient client,System.IO.Stream stream , File file)
        {
            HttpClient hc = new HttpClient(new WebRequestHandler { AllowAutoRedirect = true });
            string putLocation = GetUri(client, file);
            putLocation += "&op=" + "CREATE";
            //var resp = await hc.PutAsync(uri, null).ConfigureAwait(false);
            //var putLocation = resp.Headers.Location;
            StreamContent sc = new StreamContent(stream);
            var resp2 = await hc.PutAsync(putLocation, sc).ConfigureAwait(false);
            resp2.EnsureSuccessStatusCode();
            return resp2.Headers.Location.ToString();
        }

        public static Task<System.IO.Stream> OpenFile(this WebHDFSClient client, File file, int offset, int length)
        {
            if (length > file.Length)
            {
                length = (int)file.Length;
            }
            //var contentTask = client.OpenFile("/user/org1/projects/myproject/mybucket635150977324454665/object635150977364825853").Result;
            //var responseStream = contentTask.Content.ReadAsStreamAsync().Result;
            //System.IO.MemoryStream ms = new System.IO.MemoryStream();
            //responseStream.CopyTo(ms);
            //ms.Position = 0;
            //return null;

            string uri = GetUri(client, file);
            uri += "&op=" + "OPEN&offset=" + offset.ToString() + "&length=" + length.ToString();

            //HttpClient hc = new HttpClient(new WebRequestHandler { AllowAutoRedirect = true });
            //var resp = await hc.GetAsync(uri).ConfigureAwait(false);
            //resp.EnsureSuccessStatusCode();
            //return resp.Content.ReadAsStreamAsync().Result;

            return WebRequest.Create(uri).GetResponseAsync()
                .ContinueWith<System.IO.Stream>(
                 t =>
                 {
                     System.IO.MemoryStream ms = new System.IO.MemoryStream();
                     t.Result.GetResponseStream().CopyTo(ms);
                     ms.Position = 0;
                     return ms;
                 }
            );

        }


        public static Task<bool> Copy(this WebHDFSClient client, File sourcefile, File destinationFile)
        {
            var Logger = Inovout.ComponentRegistry.Resolve<IExtendedLogger>();
            IApplication application = ComponentRegistry.Resolve<IApplication>();

            WebRequest webRequest = WebRequest.Create(application.Settings["HadoopNameNodeHttpShellUri"].ToString());
            webRequest.Method = "POST";
            string cmds = "cmd=" +
                string.Format(application.Settings["HdfsMkdirCommand"].ToString(), destinationFile.Directory.Path)
                + "&cmd=" +
                string.Format(application.Settings["HdfsCopyCommand"].ToString(), sourcefile.Path, destinationFile.Path);
            Logger.Info("WebHdfs Copy Command:" + cmds);
            byte[] buffers = UTF8Encoding.UTF8.GetBytes(cmds);

            webRequest.GetRequestStream().Write(buffers, 0, buffers.Length);
            webRequest.ContentType = "application/x-www-form-urlencoded";

            return webRequest.GetResponseAsync().ContinueWith<bool>(
                t =>
                {
                    try
                    {
                        HttpWebResponse webResponse = (HttpWebResponse)t.Result;
                        Logger.Error("WebHdfs Copy Info:" + new System.IO.StreamReader(webResponse.GetResponseStream()).ReadToEnd());
                        if ((int)webResponse.StatusCode >= 300)
                        {
                            return false;
                        }
                        else
                        {
                            return true;
                        }
                    }
                    catch (Exception ex)
                    {
                        Logger.Error("WebHdfs Copy error.", ex);
                        return false;
                    }
                }
                );

        }
        public static async Task<DirectoryEntry> GetFileStatus(this WebHDFSClient client, File file)
        {
            var Logger = Inovout.ComponentRegistry.Resolve<IExtendedLogger>();

            string uri = GetUri(client, file);
            uri += "&op=" + "GETFILESTATUS";
            Logger.Info("GetFileStatus Uri:" + uri);
            IApplication application = ComponentRegistry.Resolve<IApplication>();
            DateTime timeoutDateTime = DateTime.Now.AddMinutes(5);
            if (application.Settings.Contains("WebHdfsTimeOut"))
            {
                timeoutDateTime = DateTime.Now.AddSeconds(double.Parse(application.Settings["WebHdfsTimeOut"].ToString()));
            }
            int webHdfsInterval = 1000;
            if (application.Settings.Contains("WebHdfsInterval"))
            {
                webHdfsInterval = int.Parse(application.Settings["WebHdfsInterval"].ToString());
            }
            //因为CP或MV有延时，故需摊牌调用。
            while (true)
            {
                try
                {
                    Logger.Info("Try GetFileStatus at " + DateTime.Now.ToString() + ":" + uri);

                    HttpWebResponse response = (HttpWebResponse)WebRequest.Create(uri).GetResponse();
                    System.IO.Stream stream = response.GetResponseStream();
                    string jsonString = new System.IO.StreamReader(stream).ReadToEnd();
                    JObject result = JObject.Parse(jsonString);
                    Logger.Info("GetFileStatus succ at " + DateTime.Now.ToString() + ":" + uri + " FileInfo:" + jsonString);
                    return new DirectoryEntry(result.Value<JObject>("FileStatus"));
                }
                catch (Exception ex)
                {
                    Logger.Error("WebHdfs GetFileStatus", ex);
                }
                if (DateTime.Now > timeoutDateTime)
                {
                    throw new Exception("Copy time out!please try after many minture.");
                }
                Thread.Sleep(webHdfsInterval);
            }
        }
    }
}
